/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.rocketmq.spring.starter.core;

import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.CONSUMEFAILED_TAG;
import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.CONSUMEFAILED_TOPIC;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Properties;

import org.apache.rocketmq.spring.starter.enums.ConsumeMode;
import org.apache.rocketmq.spring.starter.enums.SelectorType;
import org.apache.rocketmq.spring.starter.exception.ConvertMsgException;
import org.apache.rocketmq.spring.starter.msgvo.ConsumeFailedMsgVO;
import org.apache.rocketmq.spring.starter.utils.ExceptionUtil;
import org.apache.rocketmq.spring.starter.utils.IPUtil;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.batch.BatchConsumer;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import com.aliyun.openservices.ons.api.order.ConsumeOrderContext;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.MessageSelector;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.exception.MQClientException;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.fasterxml.jackson.databind.ObjectMapper;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@SuppressWarnings("WeakerAccess")
@Slf4j
public class AliyunRocketMQListenerContainer implements InitializingBean, RocketMQListenerContainer {
	/**
     * 阿里云分配的accesskey
     */
	@Setter
    private String accessKey;
    /**
     * 阿里云分配的secretKey
     */
    @Setter
    private String secretKey;
    
    @Setter
    @Getter
    private String consumerGroup;
    /**
     * 消息队列服务接入点
     */
    @Setter
    @Getter
	private String onsAddr;

    @Setter
    @Getter
    private String topic;

    @Setter
    @Getter
    private ConsumeMode consumeMode = ConsumeMode.CONCURRENTLY;

    @Setter
    @Getter
    private SelectorType selectorType = SelectorType.TAG;

    @Setter
    @Getter
    private String selectorExpress = "*";

    @Setter
    @Getter
    private MessageModel messageModel = MessageModel.CLUSTERING;

    @Setter
    @Getter
    private int consumeThreadMax = 64;

    @Getter
    @Setter
    private String charset = "UTF-8";

    @Setter
    @Getter
    private ObjectMapper objectMapper = new ObjectMapper();

    @Setter
    @Getter
    private boolean started;

    @Setter
    private RocketMQListener rocketMQListener;
    /**普通消息*/
    private Consumer consumer;
    /**顺序消息*/
    private OrderConsumer orderConsumer;
    /**批量消息*/
    private BatchConsumer batchConsumer;
    
    private Class messageType;
    /**
	 * 环境前缀
	 */
    @Setter
	private String environmentPrefix;

    @Setter
    private RocketMQTemplate rocketMQTemplate;
    
    public void setupMessageListener(RocketMQListener rocketMQListener) {
        this.rocketMQListener = rocketMQListener;
    }

    @Override
    public void destroy() {
        this.setStarted(false);
        if (Objects.nonNull(consumer)) {
            consumer.shutdown();
        }
        if (Objects.nonNull(orderConsumer)) {
        	orderConsumer.shutdown();
        }
        if (Objects.nonNull(batchConsumer)) {
        	batchConsumer.shutdown();
        }
        log.info("container destroyed, {}", this.toString());
    }

    public synchronized void start() throws MQClientException {

        if (this.isStarted()) {
            throw new IllegalStateException("container already started. " + this.toString());
        }

        initRocketMQPushConsumer();

        // parse message type
        this.messageType = getMessageType();
        log.debug("msgType: {}", messageType.getName());

        if (Objects.nonNull(consumer)) {
            consumer.start();
        }
        if (Objects.nonNull(orderConsumer)) {
        	orderConsumer.start();
        }
        if (Objects.nonNull(batchConsumer)) {
        	batchConsumer.start();
        }
        this.setStarted(true);

        log.info("started container: {}", this.toString());
    }

    public class DefaultMessageListenerConcurrently implements MessageListener {

        @SuppressWarnings("unchecked")
		public Action consume(final Message message, final ConsumeContext context){
        	Date consumeBeginTime = new Date();
            log.debug("received msg: {}", message);
            try {
                long now = consumeBeginTime.getTime();
                rocketMQListener.onMessage(doConvertMessage(message));
                long costTime = System.currentTimeMillis() - now;
                log.debug("consume {} cost: {} ms", message.getMsgID(), costTime);
            } catch (Exception e) {
                log.warn("consume message failed. message:{}", message, e);
                if(message.getTopic().equals(environmentPrefix+"_"+CONSUMEFAILED_TOPIC) && CONSUMEFAILED_TAG.equals(message.getTag())){
					log.error("消费失败的消息为“保存消费失败日志消息”，不需要记录日志,不需要重新消费，直接返回成功");
					return Action.CommitMessage;
				}
                if(e instanceof ConvertMsgException){
                	log.error("消费失败的原因为转换对象失败，需要记录日志，不需要重新消费，返回消费成功");
                	//消息消费失败，发送失败消息
                	this.sendConsumeMsgFailed(message,e,consumeBeginTime);
                	return Action.CommitMessage;
                }
            	this.sendConsumeMsgFailed(message,e,consumeBeginTime);
                return Action.ReconsumeLater;
            }
        
            return Action.CommitMessage;
        }
        /**
         * 发送消息消费失败消息
         * @param message
         * @param e
         * 2018年3月22日 zhaowg
         */
		private void sendConsumeMsgFailed(Message message, Exception e,Date consumeBeginTime) {
			log.info("消费消息失败，开始发送消费失败MQ");
			String topic = CONSUMEFAILED_TOPIC;
			String tag  = CONSUMEFAILED_TAG;
			try{
				Date consumeEndTime = new Date();
				ConsumeFailedMsgVO consumeFailedMsgVO = new ConsumeFailedMsgVO();
				consumeFailedMsgVO.setConsumeBeginTime(consumeBeginTime);
				consumeFailedMsgVO.setConsumeEndTime(consumeEndTime);
				consumeFailedMsgVO.setConsumeGroup(consumerGroup);
				consumeFailedMsgVO.setConsumeIp(IPUtil.getLocalHost());
				if(e!=null){
					String errMsg = ExceptionUtil.getTrace(e);
					if(!StringUtils.isEmpty(errMsg)){
						//最多保存1024个字符
						consumeFailedMsgVO.setCunsumerErrMsg(errMsg.substring(0, 1024));
					}
				}
				consumeFailedMsgVO.setMsg(new String(message.getBody()));
				consumeFailedMsgVO.setMsgId(message.getMsgID());
				consumeFailedMsgVO.setMsgKeys(message.getKey());
				consumeFailedMsgVO.setReconsumeTimes(message.getReconsumeTimes());
				consumeFailedMsgVO.setTag(message.getTag());
				consumeFailedMsgVO.setTopic(message.getTopic());
				rocketMQTemplate.sendOneWay(topic, tag, consumeFailedMsgVO);
				log.info("发送消息消费失败MQ成功");
			}catch(Exception e1){
				log.info("发送消息消费失败MQ异常",e);
			}
			
		}
    }

    public class DefaultMessageListenerOrderly implements MessageOrderListener {

		@Override
		public OrderAction consume(Message message, ConsumeOrderContext context) {
			log.debug("received msg: {}", message);
            try {
                long now = System.currentTimeMillis();
                rocketMQListener.onMessage(doConvertMessage(message));
                long costTime = System.currentTimeMillis() - now;
                log.info("consume {} cost: {} ms", message.getMsgID(), costTime);
            } catch (Exception e) {
                log.warn("consume message failed. message:{}", message, e);
                return OrderAction.Suspend;
            }
			return OrderAction.Success;
		}
    }
    
    public class DefaultMessageListenerBatchs  implements BatchMessageListener{

		@Override
		public Action consume(List<Message> messages, ConsumeContext context) {
			for (Message message : messages) {
				Date consumeBeginTime = new Date();
	            log.debug("received msg: {}", message);
	            try {
	                long now = consumeBeginTime.getTime();
	                rocketMQListener.onMessage(doConvertMessage(message));
	                long costTime = System.currentTimeMillis() - now;
	                log.debug("consume {} cost: {} ms", message.getMsgID(), costTime);
	            } catch (Exception e) {
	                log.warn("consume message failed. message:{}", message, e);
	                if(message.getTopic().equals(environmentPrefix+"_"+CONSUMEFAILED_TOPIC) && CONSUMEFAILED_TAG.equals(message.getTag())){
						log.error("消费失败的消息为“保存消费失败日志消息”，不需要记录日志,不需要重新消费，直接返回成功");
						continue;
					}
	                if(e instanceof ConvertMsgException){
	                	log.error("消费失败的原因为转换对象失败，需要记录日志，不需要重新消费，返回消费成功");
	                	//消息消费失败，发送失败消息
	                	this.sendConsumeMsgFailed(message,e,consumeBeginTime);
	                	continue;
	                }
	            	this.sendConsumeMsgFailed(message,e,consumeBeginTime);
	                return Action.ReconsumeLater;
	            }
			}
			return Action.CommitMessage;
		}
		
		/**
         * 发送消息消费失败消息
         * @param message
         * @param e
         * 2018年3月22日 zhaowg
         */
		private void sendConsumeMsgFailed(Message message, Exception e,Date consumeBeginTime) {
			log.info("消费消息失败，开始发送消费失败MQ");
			String topic = environmentPrefix+"_"+CONSUMEFAILED_TOPIC;
			String tag  = CONSUMEFAILED_TAG;
			try{
				Date consumeEndTime = new Date();
				ConsumeFailedMsgVO consumeFailedMsgVO = new ConsumeFailedMsgVO();
				consumeFailedMsgVO.setConsumeBeginTime(consumeBeginTime);
				consumeFailedMsgVO.setConsumeEndTime(consumeEndTime);
				consumeFailedMsgVO.setConsumeGroup(consumerGroup);
				consumeFailedMsgVO.setConsumeIp(IPUtil.getLocalHost());
				if(e!=null){
					String errMsg = ExceptionUtil.getTrace(e);
					if(!StringUtils.isEmpty(errMsg)){
						//最多保存1024个字符
						consumeFailedMsgVO.setCunsumerErrMsg(errMsg.substring(0, 1024));
					}
				}
				consumeFailedMsgVO.setMsg(new String(message.getBody()));
				consumeFailedMsgVO.setMsgId(message.getMsgID());
				consumeFailedMsgVO.setMsgKeys(message.getKey());
				consumeFailedMsgVO.setReconsumeTimes(message.getReconsumeTimes());
				consumeFailedMsgVO.setTag(message.getTag());
				consumeFailedMsgVO.setTopic(message.getTopic());
				rocketMQTemplate.sendOneWay(topic, tag, consumeFailedMsgVO);
				log.info("发送消息消费失败MQ成功");
			}catch(Exception e1){
				log.info("发送消息消费失败MQ异常",e);
			}
			
		}
    }
    @Override
    public void afterPropertiesSet() throws Exception {
        start();
    }


    @SuppressWarnings("unchecked")
    private Object doConvertMessage(Message message) {
        if (Objects.equals(messageType, Message.class)) {
            return message;
        } else {
            String str = new String(message.getBody(), Charset.forName(charset));
            if (Objects.equals(messageType, String.class)) {
                return str;
            } else {
                // if msgType not string, use objectMapper change it.
                try {
                    return objectMapper.readValue(str, messageType);
                } catch (Exception e) {
                    log.info("convert failed. str:{}, msgType:{}", str, messageType);
                    throw new ConvertMsgException("cannot convert message to " + messageType, e);
                }
            }
        }
    }

    private Class getMessageType() {
        Type[] interfaces = rocketMQListener.getClass().getGenericInterfaces();
        if (Objects.nonNull(interfaces)) {
            for (Type type : interfaces) {
                if (type instanceof ParameterizedType) {
                    ParameterizedType parameterizedType = (ParameterizedType) type;
                    if (Objects.equals(parameterizedType.getRawType(), RocketMQListener.class)) {
                        Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
                        if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
                            return (Class) actualTypeArguments[0];
                        } else {
                            return Object.class;
                        }
                    }
                }
            }

            return Object.class;
        } else {
            return Object.class;
        }
    }

    private void initRocketMQPushConsumer() throws MQClientException {

        Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
        Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
        Assert.notNull(onsAddr, "Property 'nameServer' is required");
        Assert.notNull(topic, "Property 'topic' is required");

        Properties consumerProperties = new Properties();
        consumerProperties.setProperty(PropertyKeyConst.ConsumerId, consumerGroup);
        consumerProperties.setProperty(PropertyKeyConst.AccessKey, accessKey);
        consumerProperties.setProperty(PropertyKeyConst.SecretKey, secretKey);
        consumerProperties.setProperty(PropertyKeyConst.ONSAddr, onsAddr);
        consumerProperties.setProperty(PropertyKeyConst.ConsumeThreadNums, consumeThreadMax+"");
        consumerProperties.setProperty(PropertyKeyConst.MessageModel, messageModel.getModeCN());
        //允许用户自己设置该consumer的一些配置
        if (rocketMQListener instanceof AliyunRocketMQPushConsumerLifecycleListener) {
        	((AliyunRocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumerProperties);
        }
        switch (consumeMode) {
        case ORDERLY://顺序消息
        	orderConsumer = ONSFactory.createOrderedConsumer(consumerProperties);
        	if(selectorType == SelectorType.TAG){
        		orderConsumer.subscribe(topic, selectorExpress, new DefaultMessageListenerOrderly());
//        	}else if(selectorType == SelectorType.SQL92){
//        		orderConsumer.subscribe(topic, MessageSelector.bySql(selectorExpress), new DefaultMessageListenerOrderly());
        	}
            break;
        case CONCURRENTLY://普通消息
        	consumer = ONSFactory.createConsumer(consumerProperties);
        	if(selectorType == SelectorType.TAG){
        		consumer.subscribe(topic, selectorExpress, new DefaultMessageListenerConcurrently());
//        	}else if(selectorType == SelectorType.SQL92){
//        		consumer.subscribe(topic, MessageSelector.bySql(selectorExpress), new DefaultMessageListenerConcurrently());
        	}
            break;
        case BATCH://批量消息
        	batchConsumer = ONSFactory.createBatchConsumer(consumerProperties);
        	batchConsumer.subscribe(topic, selectorExpress, new DefaultMessageListenerBatchs());
        	break;
        default:
            throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
        }
        
    }

}
